package com.zn.websocket.group;

import com.zn.util.Constant;
import com.zn.util.DateUtils;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 连接通道管理类
 * @author yinjiawei
 */
public class ChannelGroups {
    private static final Logger LOGGER = LoggerFactory.getLogger(ChannelGroups.class);
    // 所以已经连接的WebSocket Channel 都聚集到一起
    private static final ChannelGroup CHANNELS = new DefaultChannelGroup("ChannelGroups", GlobalEventExecutor.INSTANCE);
    private static final ConcurrentHashMap<String, Channel> CHANNEL_MAP = new ConcurrentHashMap<>();
    //所有动态Channel定义
    private static final ConcurrentHashMap<String, ConcurrentHashMap<String, Channel>> CHANNEL_MANAGE = new ConcurrentHashMap<>();
    /**
     * 预加载类
     * @author yinjiawei
     * @time 12/19/17 17:17
     */
    public static void preLoad() {}

    /**
     * 增加元素
     * @author yinjiawei
     * @time 12/19/17 17:18
     * @param channel
     */
    public static void add (Channel channel) {
        CHANNELS.add(channel);
        LOGGER.warn("add channel size :" + String.valueOf(CHANNELS.size()));
    }

    /**
     * 广播消息
     * @author yinjiawei
     * @time 12/19/17 17:18
     * @param msg
     */
    public static void broadcast(Object msg) {
        CHANNELS.writeAndFlush(msg);
    }

    /**
     * 根据手机号发送给用户
     * @author yinjiawei
     */
    public static void writeByTel(String tel, TextWebSocketFrame frame) {
        Channel channel = CHANNEL_MAP.get(tel);
        if (channel == null)
            return;
        if (channel.isActive()) {
            channel.writeAndFlush(frame);
            return ;
        }
        CHANNEL_MAP.remove(tel);
    }

    /**
     * 根据uuid发送给用户
     * @author yinjiawei
     */
    public static void writeByUuid(String uuid, TextWebSocketFrame frame) {
        Channel channel = CHANNEL_MAP.get(uuid);
        if (channel == null)
            return;
        if (channel.isActive()) {
            LOGGER.info("ChannelGroups:....." + uuid);
            channel.writeAndFlush(frame);
            return ;
        }
        CHANNEL_MAP.remove(uuid);
    }

    public static void writeByMonth(String channelName, TextWebSocketFrame frame) {
        ConcurrentHashMap<String, Channel> map = getChannelMap(channelName);
        if (map != null) {
            for (String key : map.keySet()) {
                if (map.get(key) == null)
                    return;
                if (map.get(key).isActive()) {
                    LOGGER.info("ChannelGroups:....." + key);
                    map.get(key).writeAndFlush(frame);
                    return;
                }
                getChannelMap(channelName).remove(key);
            }
        }
    }

    /**
     * 根据uuid发送给用户
     * @author yinjiawei
     */
    public static void writeByUuid(String uuid, BinaryWebSocketFrame frame) {
        Channel channel = CHANNEL_MAP.get(uuid);
        if (channel == null)
            return;
        if (channel.isActive()) {
            LOGGER.info("ChannelGroups:....."+uuid);
            channel.writeAndFlush(frame.retainedDuplicate());
            return ;
        }
        CHANNEL_MAP.remove(uuid);
    }

    public static void writeByMonth(String channelName, BinaryWebSocketFrame frame) {
        ConcurrentHashMap<String, Channel> mapYm = getChannelMap(channelName);
        ConcurrentHashMap<String, Channel> mapAll = CHANNEL_MANAGE.get("CHANNEL_MAP_MD_ALL");
        List<ConcurrentHashMap<String, Channel>> list = new ArrayList<>();
        if(mapYm != null){
            list.add(mapYm);
        }
        list.add(mapAll);
        for (ConcurrentHashMap<String, Channel> map : list) {
            if (map != null) {
                for (String key : map.keySet()) {
                    Channel channel=map.get(key);
                    if (channel!=null && channel.isActive()) {
                        LOGGER.info("ChannelGroups:....." + key);
                        channel.writeAndFlush(frame.retainedDuplicate());
                    }
                }
            }
        }
        return;
    }

    /**
     * 推送50ETF行情所有通道
     * @param frame
     */
    public static void writeByEtf( BinaryWebSocketFrame frame) {
        for (String key : CHANNEL_MANAGE.keySet()){
            ConcurrentHashMap<String, Channel> map = CHANNEL_MANAGE.get(key);
            if (map.size() > 0) {
                for (String UuidKey : map.keySet()) {
                    Channel channel=map.get(UuidKey);
                    if (channel != null && channel.isActive()) {
                        LOGGER.info("ChannelGroups:....." + UuidKey);
                        channel.writeAndFlush(frame.retainedDuplicate());
                    }else{
                        map.remove(UuidKey);
                    }
                }
            }
        }
        return;
    }

    public static void writeByChannelName(String channelName,BinaryWebSocketFrame frame){
        ConcurrentHashMap<String, Channel> map = CHANNEL_MANAGE.get(channelName);
        if (map.size() > 0) {
            for (String UuidKey : map.keySet()) {
                Channel channel=map.get(UuidKey);
                if (channel != null && channel.isActive()) {
                    channel.writeAndFlush(frame.retainedDuplicate());
                }else{
                    map.remove(UuidKey);
                }
            }
        }
    }


    /**
     * 刷新
     * @author yinjiawei
     */
    public static void flush() {
        CHANNELS.flush();
    }
    
    /**
     * 丢弃
     * @author yinjiawei
     */
    public static void discard(Channel channel) {
        CHANNELS.remove(channel);
        LOGGER.warn("remove from :" + String.valueOf(CHANNELS.size()));
    }

    /**
     * 断开连接
     * @author yinjiawei
     */
    public static void disConnect () {
        CHANNELS.disconnect();
    }

    /**
     * 判断是否包含
     * @author yinjiawei
     */
    public static boolean contains (Channel channel) {
        return CHANNELS.contains(channel);
    }

    /**
     * 将用户token绑定到对应的通道上
     * @author yinjiawei
     */
    public static boolean bindChannel(String uuid, Channel channel) {
        if (contains(channel) && channel != null && channel.isActive()) {
            // 如果原来有 则替换原来绑定的
            CHANNEL_MAP.put(uuid, channel);
            return true;
        }
        return false;
    }

    /**
     * 将用户token绑定到对应的相对应的通道上
     * @author yinjiawei
     */
    public static boolean bindChannel(String uuid, String channelName, Channel channel) {
        // 如果原来有 则删除
        if (contains(channel) && channel != null && channel.isActive()){
            removeUuidChannel(uuid);
        }
        if (channel != null && channel.isActive()) {

            if (!checkChannelMap(channelName)){
                addChannelMap(channelName);
            }
            getChannelMap(channelName).put(uuid, channel);
            return true;
        }
        return false;
    }

    /**
     * 新增通道集合
     * @param channelName
     */
    public static void addChannelMap(String channelName){
        CHANNEL_MANAGE.put("CHANNEL_MAP_" + channelName, new ConcurrentHashMap<String, Channel>());
    }

    /**
     * 删除通道集合
     * @param channelName
     */
    public static void removeChannelMap(String channelName){
        CHANNEL_MANAGE.remove("CHANNEL_MAP_"+channelName);
    }

    /**
     * 判断通道是否存在
     * @param channelName
     * @return
     */
    public static boolean checkChannelMap(String channelName){
        return CHANNEL_MANAGE.containsKey("CHANNEL_MAP_" + channelName);
    }

    /**
     * 获取当前通道集合
     * @param channelName
     * @return
     */
    public static ConcurrentHashMap<String, Channel> getChannelMap(String channelName){
        return CHANNEL_MANAGE.get("CHANNEL_MAP_"+channelName);
    }

    /**
     * 清除其它渠道中推送
     * @param uuid
     */
    public static void removeUuidChannel(String uuid){
        for (String key : CHANNEL_MANAGE.keySet()){
            if (CHANNEL_MANAGE.get(key).containsKey(uuid)){
                CHANNEL_MANAGE.get(key).remove(uuid);
            }
        }
    }

    public static void main(String[] args) {
//        ConcurrentHashMap<String, String> a = new ConcurrentHashMap<>();
//        a.put("aaa", "abc");
//        LOGGER.warn(a.put("aaa", "abc"));
        List<String> stringList = new ArrayList<>();
        ConcurrentHashMap<String, Channel> concurrentHashMap = new ConcurrentHashMap<>();
        stringList.add("510050");
        stringList.add("201811");
        stringList.add("201812");
        stringList.add("201903");
        stringList.add("201906");
        for (int i = 0; i < stringList.size()-2; i++){
            CHANNEL_MANAGE.put(stringList.get(i), concurrentHashMap);
        }
        System.out.println(CHANNEL_MANAGE.size());
        CHANNEL_MANAGE.clear();
        for (String str : stringList){
            CHANNEL_MANAGE.put(str, concurrentHashMap);
        }
        System.out.println(CHANNEL_MANAGE.size());


//        Map<String,String> map=new HashMap<String,String>();
//        map.put("aaaaa", "bbbb");
//        String cc=map.get("cccc");
//        System.err.println("cc-------------->"+cc);

    }

    public static void Initialization(List<String> stringList){
        CHANNEL_MANAGE.clear();
        for (String str : stringList){
            ConcurrentHashMap<String, Channel> concurrentHashMap = new ConcurrentHashMap<>();
            CHANNEL_MANAGE.put("CHANNEL_MAP_"+str, concurrentHashMap);
        }
        ConcurrentHashMap<String, Channel> concurrentHashMap = new ConcurrentHashMap<>();
        /**持仓中订阅**/
        CHANNEL_MANAGE.put("CHANNEL_MAP_MD_ALL", concurrentHashMap);
        /**现手中订阅**/
        CHANNEL_MANAGE.put(Constant.CHANNEL_MAP_NOW_VOLUME, concurrentHashMap);
        LOGGER.info(DateUtils.getDate(new Date())+"成功初始化通道共"+CHANNEL_MANAGE.size()+"个");
    }

    /**
     * ping 客户端
     * @author yinjiawei
     */
    public static void ping() {
        CHANNELS.stream().forEach((channel -> {
            channel.writeAndFlush(new PingWebSocketFrame());
        }));
    }
}
